// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#include <gtest/gtest.h>

#include <pollux/common/base/pollux_exception.h>
#include <pollux/common/memory/memory.h>
#include <pollux/common/testutil/optional_empty.h>
#include <pollux/vector/vector_maker.h>
#include <pollux/testing/vector/vector_test_base.h>

using namespace kumo::pollux;
using kumo::pollux::VectorMaker;

class VectorMakerTest : public ::testing::Test {
 protected:
  static void SetUpTestCase() {
    memory::MemoryManager::testingSetInstance({});
  }

  std::shared_ptr<memory::MemoryPool> pool_{
      memory::memoryManager()->addLeafPool()};
  VectorMaker maker_{pool_.get()};
};

TEST_F(VectorMakerTest, flat_vector) {
  std::vector<int64_t> data = {0, 1, 2, 3, 1024, -123456, -99, -999, -1};
  auto flat_vector = maker_.flat_vector(data);

  EXPECT_EQ(data.size(), flat_vector->size());
  EXPECT_FALSE(flat_vector->may_have_nulls());
  EXPECT_EQ(0, flat_vector->get_null_count().value());
  EXPECT_FALSE(flat_vector->isSorted().value());
  EXPECT_EQ(9, flat_vector->get_distinct_value_count().value());
  EXPECT_EQ(-123456, flat_vector->getMin().value());
  EXPECT_EQ(1024, flat_vector->getMax().value());

  for (vector_size_t i = 0; i < data.size(); i++) {
    EXPECT_EQ(data[i], flat_vector->value_at(i));
  }
}

TEST_F(VectorMakerTest, nullableFlatVector) {
  std::vector<std::optional<int64_t>> data = {
      0, 1, std::nullopt, 3, 1024, -123456, -99, -999, std::nullopt, -1};

  auto flat_vector = maker_.flat_vector_nullable<int64_t>(data);

  EXPECT_EQ(data.size(), flat_vector->size());
  EXPECT_TRUE(flat_vector->may_have_nulls());
  EXPECT_EQ(2, flat_vector->get_null_count().value());
  EXPECT_FALSE(flat_vector->isSorted().value());
  EXPECT_EQ(8, flat_vector->get_distinct_value_count().value());
  EXPECT_EQ(-123456, flat_vector->getMin().value());
  EXPECT_EQ(1024, flat_vector->getMax().value());

  for (vector_size_t i = 0; i < data.size(); i++) {
    if (data[i] == std::nullopt) {
      EXPECT_TRUE(flat_vector->is_null_at(i));
    } else {
      EXPECT_FALSE(flat_vector->is_null_at(i));
      EXPECT_EQ(*data[i], flat_vector->value_at(i));
    }
  }
}

TEST_F(VectorMakerTest, flatVectorString) {
  std::vector<std::string> data = {
      "hello",
      "world",
      "afe@41 135$2@%",
      "",
      "end",
  };

  auto flat_vector = maker_.flat_vector(data);

  EXPECT_EQ(data.size(), flat_vector->size());
  EXPECT_FALSE(flat_vector->may_have_nulls());
  EXPECT_EQ(0, flat_vector->get_null_count().value());
  EXPECT_FALSE(flat_vector->isSorted().value());
  EXPECT_EQ(5, flat_vector->get_distinct_value_count().value());
  EXPECT_EQ(""_sv, flat_vector->getMin().value());
  EXPECT_EQ(StringView("world"), flat_vector->getMax().value());

  for (vector_size_t i = 0; i < data.size(); i++) {
    EXPECT_EQ(data[i], std::string(flat_vector->value_at(i)));
  }
}

TEST_F(VectorMakerTest, flatVectorStringTypes) {
  auto validate = [&](const FlatVectorPtr<StringView>& input) {
    ASSERT_NE(nullptr, input);
    EXPECT_EQ("hello"_sv, input->value_at(0));
    EXPECT_EQ("world"_sv, input->value_at(1));
  };

  // char*
  validate(maker_.flat_vector({"hello", "world"}));

  // std::string
  validate(maker_.flat_vector({std::string("hello"), std::string("world")}));

  // StringView
  validate(maker_.flat_vector({"hello"_sv, "world"_sv}));

  // std::string_view
  validate(maker_.flat_vector(
      {std::string_view("hello"), std::string_view("world")}));
}

TEST_F(VectorMakerTest, flatVectorStringNullableTypes) {
  auto validate = [&](const FlatVectorPtr<StringView>& input) {
    ASSERT_NE(nullptr, input);
    EXPECT_EQ("hello"_sv, input->value_at(0));
    EXPECT_TRUE(input->is_null_at(1));
    EXPECT_EQ("world"_sv, input->value_at(2));
  };

  // Compilers can't infer dependent template types, so we either need to
  // explicitly specify the template type, of fully declare the vector type:

  // char*
  validate(
      maker_.flat_vector_nullable<const char*>({"hello", std::nullopt, "world"}));

  // std::string
  validate(maker_.flat_vector_nullable(std::vector<std::optional<std::string>>(
      {"hello", std::nullopt, "world"})));

  // StringView
  validate(
      maker_.flat_vector_nullable<StringView>({"hello", std::nullopt, "world"}));

  // std::string_view
  validate(
      maker_.flat_vector_nullable(std::vector<std::optional<std::string_view>>(
          {"hello", std::nullopt, "world"})));
}

TEST_F(VectorMakerTest, nullableFlatVectorString) {
  std::vector<std::optional<std::string>> data = {
      "hello",
      "world",
      std::nullopt,
      "",
      "end",
  };

  auto flat_vector = maker_.flat_vector_nullable(data);

  EXPECT_EQ(data.size(), flat_vector->size());
  EXPECT_TRUE(flat_vector->may_have_nulls());
  EXPECT_EQ(1, flat_vector->get_null_count().value());
  EXPECT_FALSE(flat_vector->isSorted().value());
  EXPECT_EQ(4, flat_vector->get_distinct_value_count().value());
  EXPECT_EQ(""_sv, flat_vector->getMin().value());
  EXPECT_EQ("world"_sv, flat_vector->getMax().value());

  for (vector_size_t i = 0; i < data.size(); i++) {
    if (data[i] == std::nullopt) {
      EXPECT_TRUE(flat_vector->is_null_at(i));
    } else {
      EXPECT_FALSE(flat_vector->is_null_at(i));
      EXPECT_EQ(*data[i], std::string(flat_vector->value_at(i)));
    }
  }
}

TEST_F(VectorMakerTest, nullableFlatVectorBool) {
  std::vector<std::optional<bool>> data = {
      true,
      false,
      std::nullopt,
      false,
      true,
  };

  auto flat_vector = maker_.flat_vector_nullable(data);

  EXPECT_EQ(data.size(), flat_vector->size());
  EXPECT_TRUE(flat_vector->may_have_nulls());
  EXPECT_EQ(1, flat_vector->get_null_count().value());
  EXPECT_FALSE(flat_vector->isSorted().value());
  EXPECT_EQ(2, flat_vector->get_distinct_value_count().value());
  EXPECT_EQ(false, flat_vector->getMin().value());
  EXPECT_EQ(true, flat_vector->getMax().value());

  for (vector_size_t i = 0; i < data.size(); i++) {
    if (data[i] == std::nullopt) {
      EXPECT_TRUE(flat_vector->is_null_at(i));
    } else {
      EXPECT_FALSE(flat_vector->is_null_at(i));
      EXPECT_EQ(*data[i], flat_vector->value_at(i));
    }
  }
}

TEST_F(VectorMakerTest, array_vector) {
  std::vector<std::vector<int64_t>> data = {
      {0, 0},
      {1, 2, 3},
      {1024, -123, -99, -999},
      {},
      {-1},
  };

  auto array_vector = maker_.array_vector<int64_t>(data);

  EXPECT_FALSE(array_vector->may_have_nulls());

  // Validate array sizes and offsets.
  EXPECT_EQ(5, array_vector->size());

  EXPECT_EQ(2, array_vector->sizeAt(0));
  EXPECT_EQ(3, array_vector->sizeAt(1));
  EXPECT_EQ(4, array_vector->sizeAt(2));
  EXPECT_EQ(0, array_vector->sizeAt(3));
  EXPECT_EQ(1, array_vector->sizeAt(4));

  EXPECT_EQ(0, array_vector->offsetAt(0));
  EXPECT_EQ(2, array_vector->offsetAt(1));
  EXPECT_EQ(5, array_vector->offsetAt(2));
  EXPECT_EQ(9, array_vector->offsetAt(3));
  EXPECT_EQ(9, array_vector->offsetAt(4));

  // Validate actual vector elements.
  auto* elementsVector = array_vector->elements()->as_flat_vector<int64_t>();

  EXPECT_FALSE(elementsVector->may_have_nulls());

  vector_size_t idx = 0;
  for (const auto& item : data) {
    for (auto i : item) {
      EXPECT_EQ(i, elementsVector->value_at(idx++));
    }
  }
}

TEST_F(VectorMakerTest, arrayVectorString) {
  auto array_vector = maker_.array_vector<std::string>({
      {"a", "abc", "Somewhat long", "test"},
      {"b", "Another long string"},
  });

  EXPECT_FALSE(array_vector->may_have_nulls());

  // Validate array sizes and offsets.
  EXPECT_EQ(2, array_vector->size());

  EXPECT_EQ(4, array_vector->sizeAt(0));
  EXPECT_EQ(2, array_vector->sizeAt(1));

  EXPECT_EQ(0, array_vector->offsetAt(0));
  EXPECT_EQ(4, array_vector->offsetAt(1));

  // Validate actual vector elements.
  auto* elementsVector = array_vector->elements()->as_flat_vector<StringView>();

  EXPECT_FALSE(elementsVector->may_have_nulls());

  EXPECT_EQ("a", elementsVector->value_at(0).str());
  EXPECT_EQ("abc", elementsVector->value_at(1).str());
  EXPECT_EQ("Somewhat long", elementsVector->value_at(2).str());
  EXPECT_EQ("test", elementsVector->value_at(3).str());

  EXPECT_EQ("b", elementsVector->value_at(4).str());
  EXPECT_EQ("Another long string", elementsVector->value_at(5).str());
}

TEST_F(VectorMakerTest, nullableArrayVector) {
  auto O = [](std::vector<std::optional<int64_t>> data) {
    return std::make_optional(data);
  };

  std::vector<std::optional<std::vector<std::optional<int64_t>>>> data = {
      O({std::nullopt}),
      O({1, 2, 3}),
      O({1024, std::nullopt, -99, -999}),
      O({}),
      O({std::nullopt, -1}),
  };

  auto array_vector = maker_.array_vector_nullable<int64_t>(data);

  // Validate array sizes and offsets.
  EXPECT_EQ(5, array_vector->size());

  EXPECT_EQ(1, array_vector->sizeAt(0));
  EXPECT_EQ(3, array_vector->sizeAt(1));
  EXPECT_EQ(4, array_vector->sizeAt(2));
  EXPECT_EQ(0, array_vector->sizeAt(3));
  EXPECT_EQ(2, array_vector->sizeAt(4));

  EXPECT_EQ(0, array_vector->offsetAt(0));
  EXPECT_EQ(1, array_vector->offsetAt(1));
  EXPECT_EQ(4, array_vector->offsetAt(2));
  EXPECT_EQ(8, array_vector->offsetAt(3));
  EXPECT_EQ(8, array_vector->offsetAt(4));

  // Validate actual vector elements.
  auto* elementsVector = array_vector->elements()->as_flat_vector<int64_t>();

  EXPECT_TRUE(elementsVector->may_have_nulls());

  vector_size_t idx = 0;
  for (const auto& item : data) {
    for (auto i : item.value()) {
      if (i == std::nullopt) {
        EXPECT_TRUE(elementsVector->is_null_at(idx));
      } else {
        EXPECT_FALSE(elementsVector->is_null_at(idx));
        EXPECT_EQ(i, elementsVector->value_at(idx));
      }
      ++idx;
    }
  }
}

TEST_F(VectorMakerTest, arrayVectorWithNulls) {
  std::vector<std::vector<int64_t>> data = {
      {0, 0},
      {},
      {1024, -123, -99, -999},
      {},
      {-1},
  };
  std::vector<bool> nulls = {false, true, false, false, false};

  //  auto array_vector = maker_.array_vector<int64_t>(data);
  auto array_vector = maker_.array_vector<int64_t>(
      data.size(),
      [data](vector_size_t i) -> vector_size_t {
        return data[i].size();
      }, // sizeAt
      [data](vector_size_t i, vector_size_t j) -> int64_t {
        return data[i][j];
      }, // value_at
      [nulls](vector_size_t i) -> bool { return nulls[i]; } // nullAt
  );

  EXPECT_TRUE(array_vector->may_have_nulls());

  // Validate array sizes and offsets.
  EXPECT_EQ(5, array_vector->size());

  EXPECT_EQ(2, array_vector->sizeAt(0));
  EXPECT_EQ(0, array_vector->sizeAt(1));
  EXPECT_EQ(4, array_vector->sizeAt(2));
  EXPECT_EQ(0, array_vector->sizeAt(3));
  EXPECT_EQ(1, array_vector->sizeAt(4));

  EXPECT_EQ(0, array_vector->offsetAt(0));
  EXPECT_EQ(0, array_vector->offsetAt(1));
  EXPECT_EQ(2, array_vector->offsetAt(2));
  EXPECT_EQ(6, array_vector->offsetAt(3));
  EXPECT_EQ(6, array_vector->offsetAt(4));

  // Validate array null entries
  EXPECT_TRUE(array_vector->may_have_nulls());
  EXPECT_EQ(1, array_vector->get_null_count().value());
  for (int i = 0; i < nulls.size(); i++) {
    EXPECT_EQ(array_vector->is_null_at(i), nulls[i]);
  }

  // Validate actual vector elements.
  auto* elementsVector = array_vector->elements()->as_flat_vector<int64_t>();
  EXPECT_FALSE(elementsVector->may_have_nulls());

  vector_size_t idx = 0;
  for (const auto& item : data) {
    for (auto i : item) {
      EXPECT_EQ(i, elementsVector->value_at(idx++));
    }
  }
}

TEST_F(VectorMakerTest, array_vector_from_json) {
  auto array_vector = maker_.array_vector_from_json<int32_t>({
      "null",
      "[]",
      "[1, 2, 3]",
      "[1, 2, null, 4]",
      "[null, null]",
  });

  EXPECT_EQ(*array_vector->type(), *ARRAY(INTEGER()));
  EXPECT_TRUE(array_vector->may_have_nulls());

  EXPECT_EQ(5, array_vector->size());

  auto* elements = array_vector->elements()->as<FlatVector<int32_t>>();
  EXPECT_TRUE(elements->may_have_nulls());

  // Null array.
  EXPECT_TRUE(array_vector->is_null_at(0));

  // Empty array.
  EXPECT_FALSE(array_vector->is_null_at(1));
  EXPECT_EQ(0, array_vector->sizeAt(1));

  // [1, 2, 3].
  EXPECT_FALSE(array_vector->is_null_at(2));
  EXPECT_EQ(3, array_vector->sizeAt(2));
  EXPECT_EQ(0, array_vector->offsetAt(2));

  for (auto i = 0; i < 3; ++i) {
    EXPECT_FALSE(elements->is_null_at(i));
    EXPECT_EQ(i + 1, elements->value_at(i));
  }

  // [1, 2, null, 4].
  EXPECT_FALSE(array_vector->is_null_at(3));
  EXPECT_EQ(4, array_vector->sizeAt(3));
  EXPECT_EQ(3, array_vector->offsetAt(3));

  for (auto i = 0; i < 4; ++i) {
    auto index = 3 + i;
    if (i == 2) {
      EXPECT_TRUE(elements->is_null_at(index));
    } else {
      EXPECT_FALSE(elements->is_null_at(index));
      EXPECT_EQ(i + 1, elements->value_at(index));
    }
  }

  // [null, null].
  EXPECT_FALSE(array_vector->is_null_at(4));
  EXPECT_EQ(2, array_vector->sizeAt(4));
  EXPECT_EQ(7, array_vector->offsetAt(4));

  for (auto i = 0; i < 2; ++i) {
    EXPECT_TRUE(elements->is_null_at(7 + i));
  }

  auto dates = maker_.array_vector_from_json<int32_t>(
      {
          "null",
          "[]",
          "[1, 2, 3]",
          "[1, 2, null, 4]",
          "[null, null]",
      },
      ARRAY(DATE()));

  EXPECT_EQ(*dates->type(), *ARRAY(DATE()));
  EXPECT_TRUE(dates->may_have_nulls());
  EXPECT_EQ(5, dates->size());

  for (auto i = 0; i < 5; ++i) {
    EXPECT_TRUE(dates->equal_value_at(array_vector.get(), i, i));
  }
}

TEST_F(VectorMakerTest, nested_array_vector_from_json) {
  auto array_vector = maker_.nested_array_vector_from_json<int32_t>(
      {"[[1, 2], [2, 3, 4], [null, 7]]",
       "[[1, 3, 7, 9], []]",
       "[]",
       "[null]",
       "[[1, 2, null], [], null]",
       "null",
       "[[1,2,3],[],[4,5]]"});

  VectorPtr expectedVector = maker_.array_vector_nullable<int32_t>({
      {{1, 2}},
      {{2, 3, 4}},
      {{std::nullopt, 7}},
      {{1, 3, 7, 9}},
      common::testutil::optionalEmpty,
      {{std::nullopt}},
      {{1, 2, std::nullopt}},
      common::testutil::optionalEmpty,
      std::nullopt,
      {{1, 2, 3}},
      common::testutil::optionalEmpty,
      {{4, 5}},
  });

  VectorPtr elementsOfArrayVector = array_vector->elements();

  EXPECT_EQ(*array_vector->elements()->type(), *ARRAY(INTEGER()));
  EXPECT_TRUE(array_vector->elements()->may_have_nulls());

  auto rowOfArrays = array_vector->elements();

  // 7 rows in array of arrays
  EXPECT_EQ(7, array_vector->size());

  EXPECT_EQ(3, array_vector->sizeAt(0));
  EXPECT_EQ(2, array_vector->sizeAt(1));

  // Empty inner array
  EXPECT_EQ(0, array_vector->sizeAt(2));
  EXPECT_FALSE(array_vector->is_null_at(2));

  // inner array with null
  EXPECT_EQ(1, array_vector->sizeAt(3));
  EXPECT_FALSE(array_vector->is_null_at(3));

  EXPECT_EQ(3, array_vector->sizeAt(4));

  EXPECT_EQ(0, array_vector->sizeAt(5));
  EXPECT_TRUE(array_vector->is_null_at(5));

  EXPECT_EQ(3, array_vector->sizeAt(6));

  test::assertEqualVectors(expectedVector, elementsOfArrayVector);
}

TEST_F(VectorMakerTest, array_of_row_vector) {
  std::vector<std::vector<variant>> data = {
      {
          variant::row({1, "red"}),
          variant::row({2, "blue"}),
          variant::row({3, "green"}),
      },
      {},
      {
          variant::row({4, "green"}),
          variant::row({-5, "purple"}),
      },
  };

  auto array_vector = maker_.array_of_row_vector(ROW({INTEGER(), VARCHAR()}), data);

  // The arrays themselves can't be null.
  EXPECT_FALSE(array_vector->may_have_nulls());

  // Validate array sizes and offsets.
  EXPECT_EQ(3, array_vector->size());

  EXPECT_EQ(3, array_vector->sizeAt(0));
  EXPECT_EQ(0, array_vector->sizeAt(1));
  EXPECT_EQ(2, array_vector->sizeAt(2));

  EXPECT_EQ(0, array_vector->offsetAt(0));
  EXPECT_EQ(3, array_vector->offsetAt(1));
  EXPECT_EQ(3, array_vector->offsetAt(2));

  // Validate actual vector elements.
  auto* elementsVector = array_vector->elements()->as<RowVector>();

  EXPECT_FALSE(elementsVector->may_have_nulls());
  EXPECT_EQ(5, elementsVector->size());

  auto numVector = elementsVector->childAt(0)->as_flat_vector<int32_t>();
  EXPECT_EQ(5, numVector->size());

  EXPECT_EQ(1, numVector->value_at(0));
  EXPECT_EQ(2, numVector->value_at(1));
  EXPECT_EQ(3, numVector->value_at(2));
  EXPECT_EQ(4, numVector->value_at(3));
  EXPECT_EQ(-5, numVector->value_at(4));

  auto colorVector = elementsVector->childAt(1)->as_flat_vector<StringView>();
  EXPECT_EQ(5, colorVector->size());

  EXPECT_EQ("red", colorVector->value_at(0).str());
  EXPECT_EQ("blue", colorVector->value_at(1).str());
  EXPECT_EQ("green", colorVector->value_at(2).str());
  EXPECT_EQ("green", colorVector->value_at(3).str());
  EXPECT_EQ("purple", colorVector->value_at(4).str());
}

TEST_F(VectorMakerTest, arrayOfRowVectorFromTuples) {
  std::vector<std::vector<std::optional<std::tuple<int32_t, std::string>>>>
      data = {
          {{{1, "red"}}, {{2, "blue"}}, {{3, "green"}}},
          {},
          {std::nullopt},
          {{{4, "green"}}, {{-5, "purple"}}},
      };

  auto array_vector = maker_.array_of_row_vector(data, ROW({INTEGER(), VARCHAR()}));

  std::vector<vector_size_t> offsets{0, 3, 3, 4};
  auto elements = maker_.row_vector({
      maker_.flat_vector<int32_t>({1, 2, 3, 0, 4, -5}),
      maker_.flat_vector<std::string>(
          {"red", "blue", "green", "n/a", "green", "purple"}),
  });
  elements->set_null(3, true);
  auto expected = maker_.array_vector(offsets, elements);

  ASSERT_EQ(expected->size(), array_vector->size());
  // check equivalent because array_vector's row type doesn't have name for each
  // column ('', '' ..) whereas expected's row type have names ('c0', 'c1' ..)
  ASSERT_TRUE((*expected->type()).equivalent((*array_vector->type())));
  for (auto i = 0; i < expected->size(); i++) {
    ASSERT_TRUE(expected->equal_value_at(array_vector.get(), i, i));
  }
}

TEST_F(VectorMakerTest, arrayVectorUsingBaseVector) {
  auto elementsVector = maker_.flat_vector<int64_t>({1, 2, 3, 4, 5, 6});

  // Create an array vector with 2 elements per array .
  auto array_vector = maker_.array_vector({0, 2, 4}, elementsVector);

  EXPECT_EQ(array_vector->size(), 3);
  for (int i = 0; i < 3; i++) {
    EXPECT_EQ(array_vector->sizeAt(i), 2);
    EXPECT_EQ(array_vector->is_null_at(i), false);
  }

  auto rawArrayValues = array_vector->elements()->values()->as<int64_t>();
  auto baseValues = elementsVector->values()->as<int64_t>();
  EXPECT_EQ(
      memcmp(
          rawArrayValues, baseValues, elementsVector->size() * sizeof(int64_t)),
      0);

  // Create array vector with last array as null.
  auto arrayVectorWithNull =
      maker_.array_vector({0, 2, 4, 6}, elementsVector, {3});
  EXPECT_EQ(arrayVectorWithNull->is_null_at(3), true);
  EXPECT_EQ(arrayVectorWithNull->sizeAt(3), 0);
}

TEST_F(VectorMakerTest, mapVectorUsingKeyValueVectorsNoNulls) {
  auto keys = maker_.flat_vector<int32_t>({1, 2, 3, 4, 5, 6});
  auto values = maker_.flat_vector<int64_t>({7, 8, 9, 10, 11, 12});

  // Create a map vector with 2 entries per map.
  auto map_vector = maker_.map_vector({0, 2, 4}, keys, values);

  EXPECT_EQ(map_vector->size(), 3);
  for (int i = 0; i < 3; i++) {
    EXPECT_EQ(map_vector->sizeAt(i), 2);
    EXPECT_EQ(map_vector->is_null_at(i), false);
  }

  auto rawMapKeys = map_vector->mapKeys()->values()->as<int32_t>();
  auto baseKeys = keys->values()->as<int32_t>();
  EXPECT_EQ(memcmp(rawMapKeys, baseKeys, keys->size() * sizeof(int32_t)), 0);

  auto rawMapValues = map_vector->mapValues()->values()->as<int64_t>();
  auto baseValues = values->values()->as<int64_t>();
  EXPECT_EQ(
      memcmp(rawMapValues, baseValues, values->size() * sizeof(int64_t)), 0);
}

TEST_F(VectorMakerTest, mapVectorUsingKeyValueVectorsSomeNulls) {
  auto keys = maker_.flat_vector<int32_t>({1, 2, 3, 4, 5, 6});
  auto values = maker_.flat_vector<int64_t>({7, 8, 9, 10, 11, 12});

  // Create map vector with last map as null.
  auto mapVectorWithLastNull =
      maker_.map_vector({0, 2, 4, 6}, keys, values, {3});
  EXPECT_EQ(mapVectorWithLastNull->is_null_at(3), true);
  EXPECT_EQ(mapVectorWithLastNull->sizeAt(3), 0);

  // Create map vector with middle map as null.
  auto mapVectorWithMiddleNull =
      maker_.map_vector({0, 2, 2, 4, 6}, keys, values, {1});
  EXPECT_EQ(mapVectorWithMiddleNull->is_null_at(1), true);
  EXPECT_EQ(mapVectorWithMiddleNull->sizeAt(1), 0);
}

TEST_F(VectorMakerTest, mapVectorUsingKeyValueVectorsAllNulls) {
  auto keys = maker_.flat_vector<int32_t>({});
  auto values = maker_.flat_vector<int64_t>({});

  // Create map vector with last map as null.
  auto map_vector = maker_.map_vector({0, 0, 0}, keys, values, {0, 1, 2});

  EXPECT_EQ(map_vector->size(), 3);
  for (int i = 0; i < 3; i++) {
    EXPECT_EQ(map_vector->sizeAt(i), 0);
    EXPECT_EQ(map_vector->is_null_at(i), true);
  }
}

TEST_F(VectorMakerTest, mapVectorUsingKeyValueVectorsUnevenKeysValues) {
  auto keys = maker_.flat_vector<int32_t>({1, 2, 3, 4, 5, 6});
  // Create map vector with uneven keys and values, should fail.
  auto values = maker_.flat_vector<int64_t>({7, 8, 9});
  EXPECT_THROW(maker_.map_vector({0, 2, 4}, keys, values), PolluxRuntimeError);
}

TEST_F(VectorMakerTest, mapVectorUsingKeyValueVectorsNullsInvalidIndices) {
  auto keys = maker_.flat_vector<int32_t>({0, 1, 2, 3, 4, 5});
  auto values = maker_.flat_vector<int64_t>({6, 7, 8, 9, 10, 11});

  // The middle map is NULL, but according to the offsets it has size 2, this
  // should fail.
  EXPECT_THROW(
      maker_.map_vector({0, 2, 4}, keys, values, {1}), PolluxRuntimeError);
}

TEST_F(VectorMakerTest, mapVectorStringString) {
  auto map_vector = maker_.map_vector<std::string, std::string>({
      {{"a", "1"}, {"b", "2"}},
      {{"a", "This is a test"}, {"b", "This is another test"}, {"c", "test"}},
  });

  EXPECT_FALSE(map_vector->may_have_nulls());

  // Validate map sizes and offsets.
  EXPECT_EQ(2, map_vector->size());

  EXPECT_EQ(2, map_vector->sizeAt(0));
  EXPECT_EQ(3, map_vector->sizeAt(1));

  EXPECT_EQ(0, map_vector->offsetAt(0));
  EXPECT_EQ(2, map_vector->offsetAt(1));

  // Validate map keys and values.
  auto keys = map_vector->mapKeys()->as_flat_vector<StringView>();
  auto values = map_vector->mapValues()->as_flat_vector<StringView>();

  EXPECT_FALSE(keys->may_have_nulls());

  EXPECT_EQ("a", keys->value_at(0));
  EXPECT_EQ("1", values->value_at(0));

  EXPECT_EQ("b", keys->value_at(1));
  EXPECT_EQ("2", values->value_at(1));

  EXPECT_EQ("a", keys->value_at(2));
  EXPECT_EQ("This is a test", values->value_at(2));

  EXPECT_EQ("b", keys->value_at(3));
  EXPECT_EQ("This is another test", values->value_at(3));

  EXPECT_EQ("c", keys->value_at(4));
  EXPECT_EQ("test", values->value_at(4));
}

TEST_F(VectorMakerTest, map_vector_from_json) {
  auto map_vector = maker_.map_vector_from_json<int32_t, double>({
      "null",
      "{}",
      "{1: 10.1, 2: 20.2, 3: 30.3}",
      "{1: 10.1, 2: 20.2, 3: null, 4: 40.4}",
      "{1: null, 2: null}",
  });

  MapVector::canonicalize(map_vector);

  EXPECT_EQ(5, map_vector->size());
  EXPECT_EQ(*MAP(INTEGER(), DOUBLE()), *map_vector->type());
  EXPECT_TRUE(map_vector->may_have_nulls());

  auto* keys = map_vector->mapKeys()->as<SimpleVector<int32_t>>();
  auto* values = map_vector->mapValues()->as<SimpleVector<double>>();

  // Null map.
  EXPECT_TRUE(map_vector->is_null_at(0));

  // Empty map.
  EXPECT_FALSE(map_vector->is_null_at(1));
  EXPECT_EQ(0, map_vector->sizeAt(1));

  // {1: 10.1, 2: 20.2, 3: 30.3}.
  EXPECT_FALSE(map_vector->is_null_at(2));
  EXPECT_EQ(3, map_vector->sizeAt(2));
  EXPECT_EQ(0, map_vector->offsetAt(2));

  for (auto i = 0; i < 3; ++i) {
    EXPECT_FALSE(keys->is_null_at(i));
    EXPECT_EQ(i + 1, keys->value_at(i));

    EXPECT_FALSE(values->is_null_at(i));
    EXPECT_DOUBLE_EQ((i + 1) * 10.1, values->value_at(i));
  }

  // {1: 10.1, 2: 20.2, 3: null, 4: 40.4}.
  EXPECT_FALSE(map_vector->is_null_at(3));
  EXPECT_EQ(4, map_vector->sizeAt(3));
  EXPECT_EQ(3, map_vector->offsetAt(3));

  for (auto i = 0; i < 4; ++i) {
    auto index = i + 3;

    EXPECT_FALSE(keys->is_null_at(index));
    EXPECT_EQ(i + 1, keys->value_at(index));

    if (i == 2) {
      EXPECT_TRUE(values->is_null_at(index));
    } else {
      EXPECT_FALSE(values->is_null_at(index));
      EXPECT_DOUBLE_EQ((i + 1) * 10.1, values->value_at(index));
    }
  }

  // {1: null, 2: null}.
  EXPECT_FALSE(map_vector->is_null_at(4));
  EXPECT_EQ(2, map_vector->sizeAt(4));
  EXPECT_EQ(7, map_vector->offsetAt(4));

  for (auto i = 0; i < 2; ++i) {
    auto index = i + 7;

    EXPECT_FALSE(keys->is_null_at(index));
    EXPECT_EQ(i + 1, keys->value_at(index));

    EXPECT_TRUE(values->is_null_at(index));
  }
}

TEST_F(VectorMakerTest, bias_vector) {
  std::vector<std::optional<int64_t>> data = {10, 13, std::nullopt, 15, 12, 11};
  auto bias_vector = maker_.bias_vector(data);

  EXPECT_EQ(data.size(), bias_vector->size());
  EXPECT_TRUE(bias_vector->may_have_nulls());
  EXPECT_EQ(1, bias_vector->get_null_count().value());
  EXPECT_FALSE(bias_vector->isSorted().value());
  EXPECT_EQ(10, bias_vector->getMin().value());
  EXPECT_EQ(15, bias_vector->getMax().value());

  for (vector_size_t i = 0; i < data.size(); i++) {
    if (data[i] == std::nullopt) {
      EXPECT_TRUE(bias_vector->is_null_at(i));
    } else {
      EXPECT_FALSE(bias_vector->is_null_at(i));
      EXPECT_EQ(*data[i], bias_vector->value_at(i));
    }
  }
}

TEST_F(VectorMakerTest, sequence_vector) {
  std::vector<std::optional<int64_t>> data = {
      10, 10, 11, 11, std::nullopt, std::nullopt, 15, std::nullopt, 12, 12, 12};
  auto sequence_vector = maker_.sequence_vector(data);

  EXPECT_EQ(data.size(), sequence_vector->size());
  EXPECT_TRUE(sequence_vector->may_have_nulls());
  EXPECT_EQ(6, sequence_vector->numSequences());
  EXPECT_EQ(3, sequence_vector->get_null_count().value());
  EXPECT_FALSE(sequence_vector->isSorted().value());
  EXPECT_EQ(4, sequence_vector->get_distinct_value_count().value());
  EXPECT_EQ(10, sequence_vector->getMin().value());
  EXPECT_EQ(15, sequence_vector->getMax().value());

  for (vector_size_t i = 0; i < data.size(); i++) {
    if (data[i] == std::nullopt) {
      EXPECT_TRUE(sequence_vector->is_null_at(i));
    } else {
      EXPECT_FALSE(sequence_vector->is_null_at(i));
      EXPECT_EQ(*data[i], sequence_vector->value_at(i));
    }
  }
}

TEST_F(VectorMakerTest, sequenceVectorString) {
  std::vector<std::optional<StringView>> data = {
      "a"_sv,
      "a"_sv,
      "a"_sv,
      std::nullopt,
      std::nullopt,
      "b"_sv,
      std::nullopt,
      "c"_sv,
      "c"_sv,
  };
  auto sequence_vector = maker_.sequence_vector(data);

  EXPECT_EQ(data.size(), sequence_vector->size());
  EXPECT_TRUE(sequence_vector->may_have_nulls());
  EXPECT_EQ(5, sequence_vector->numSequences());
  EXPECT_EQ(3, sequence_vector->get_null_count().value());
  EXPECT_FALSE(sequence_vector->isSorted().value());
  EXPECT_EQ(3, sequence_vector->get_distinct_value_count().value());
  EXPECT_EQ("a"_sv, sequence_vector->getMin().value());
  EXPECT_EQ("c"_sv, sequence_vector->getMax().value());

  for (vector_size_t i = 0; i < data.size(); i++) {
    if (data[i] == std::nullopt) {
      EXPECT_TRUE(sequence_vector->is_null_at(i));
    } else {
      EXPECT_FALSE(sequence_vector->is_null_at(i));
      EXPECT_EQ(*data[i], sequence_vector->value_at(i));
    }
  }
}

TEST_F(VectorMakerTest, constant_vector) {
  std::vector<std::optional<int64_t>> data = {99, 99, 99};
  auto constant_vector = maker_.constant_vector(data);

  EXPECT_FALSE(constant_vector->may_have_nulls());
  EXPECT_EQ(data.size(), constant_vector->size());

  for (vector_size_t i = 0; i < data.size(); i++) {
    EXPECT_FALSE(constant_vector->is_null_at(i));
    EXPECT_EQ(*data.front(), constant_vector->value_at(i));
  }

  // Null vector.
  data = {std::nullopt, std::nullopt};
  constant_vector = maker_.constant_vector(data);

  EXPECT_TRUE(constant_vector->may_have_nulls());
  EXPECT_EQ(data.size(), constant_vector->size());

  for (vector_size_t i = 0; i < data.size(); i++) {
    EXPECT_TRUE(constant_vector->is_null_at(i));
  }
}

TEST_F(VectorMakerTest, constantRowVector) {
  auto vector = maker_.constant_row(
      ROW({INTEGER(), VARCHAR()}), variant::row({5, "orange"}), 123);

  EXPECT_EQ(vector->encoding(), VectorEncoding::Simple::CONSTANT);
  EXPECT_EQ(123, vector->size());

  auto constant_vector = vector->as<ConstantVector<ComplexType>>();
  EXPECT_EQ(0, constant_vector->index());

  auto row_vector = constant_vector->value_vector()->as<RowVector>();
  ASSERT_EQ(1, row_vector->size());

  auto numVector = row_vector->childAt(0)->as_flat_vector<int32_t>();
  ASSERT_EQ(1, numVector->size());
  ASSERT_EQ(5, numVector->value_at(0));

  auto colorVector = row_vector->childAt(1)->as_flat_vector<StringView>();
  ASSERT_EQ(1, colorVector->size());
  ASSERT_EQ("orange", colorVector->value_at(0).str());
}

TEST_F(VectorMakerTest, constantVectorErrors) {
  // Error variations.
  EXPECT_THROW(maker_.constant_vector<int64_t>({}), PolluxRuntimeError);
  EXPECT_THROW(maker_.constant_vector<int64_t>({1, 2}), PolluxRuntimeError);
  EXPECT_THROW(
      maker_.constant_vector<int64_t>({1, 2, 3, 4, 5}), PolluxRuntimeError);

  EXPECT_THROW(
      maker_.constant_vector<int64_t>({std::nullopt, 1}), PolluxRuntimeError);
  EXPECT_THROW(
      maker_.constant_vector<int64_t>({1, std::nullopt}), PolluxRuntimeError);
  EXPECT_THROW(
      maker_.constant_vector<int64_t>({1, std::nullopt, 1}), PolluxRuntimeError);
}

TEST_F(VectorMakerTest, dictionary_vector) {
  std::vector<std::optional<int64_t>> data = {
      99, 99, 11, 10, 99, std::nullopt, 10, std::nullopt};
  auto dictionary_vector = maker_.dictionary_vector(data);

  EXPECT_EQ(data.size(), dictionary_vector->size());
  EXPECT_TRUE(dictionary_vector->may_have_nulls());
  EXPECT_EQ(2, dictionary_vector->get_null_count().value());
  EXPECT_FALSE(dictionary_vector->isSorted().value());
  EXPECT_EQ(3, dictionary_vector->get_distinct_value_count().value());
  EXPECT_EQ(10, dictionary_vector->getMin().value());
  EXPECT_EQ(99, dictionary_vector->getMax().value());

  for (vector_size_t i = 0; i < data.size(); i++) {
    if (data[i] == std::nullopt) {
      EXPECT_TRUE(dictionary_vector->is_null_at(i));
    } else {
      EXPECT_FALSE(dictionary_vector->is_null_at(i));
      EXPECT_EQ(*data[i], dictionary_vector->value_at(i));
    }
  }
}

TEST_F(VectorMakerTest, isSorted) {
  // Empty and single element.
  EXPECT_TRUE(maker_.flat_vector(std::vector<int64_t>())->isSorted().value());
  EXPECT_TRUE(maker_.flat_vector(std::vector<int64_t>(10))->isSorted().value());

  // More variations and data types.
  EXPECT_TRUE(maker_.flat_vector({-1, 0, 1, 2})->isSorted().value());
  EXPECT_FALSE(maker_.flat_vector({-1, 0, 2, 1})->isSorted().value());

  EXPECT_TRUE(maker_.flat_vector({-1.9, 0.0, 9.1, 10.09})->isSorted().value());
  EXPECT_FALSE(maker_.flat_vector({-1.9, 0.0, -9.1, 10.09})->isSorted().value());

  EXPECT_TRUE(
      maker_.flat_vector({false, false, true, true})->isSorted().value());
  EXPECT_FALSE(
      maker_.flat_vector({false, false, true, false})->isSorted().value());

  // Nullable.
  EXPECT_FALSE(maker_.flat_vector_nullable<int64_t>({-1, std::nullopt, 1, 2})
                   ->isSorted()
                   .value());
  EXPECT_TRUE(
      maker_.flat_vector_nullable<int64_t>({-1, 0, 1, 2})->isSorted().value());
  EXPECT_TRUE(maker_.flat_vector_nullable<int64_t>({std::nullopt, -1, 0, 1, 2})
                  ->isSorted()
                  .value());
  EXPECT_FALSE(maker_.flat_vector_nullable<int64_t>({-1, 0, 1, 2, std::nullopt})
                   ->isSorted()
                   .value());

  // Biased.
  EXPECT_TRUE(maker_.bias_vector<int64_t>({std::nullopt, 10, 13, 13, 14, 15})
                  ->isSorted()
                  .value());
  EXPECT_FALSE(maker_.bias_vector<int64_t>({10, 13, std::nullopt, 15, 12, 11})
                   ->isSorted()
                   .value());

  // Sequence.
  EXPECT_TRUE(
      maker_.sequence_vector<int64_t>({std::nullopt, 10, 10, 11, 15, 15, 15})
          ->isSorted()
          .value());
  EXPECT_FALSE(maker_.sequence_vector<int64_t>({10, 10, std::nullopt, 9, 9, 9})
                   ->isSorted()
                   .value());
  // Dictionary.
  EXPECT_TRUE(
      maker_.dictionary_vector<int64_t>({std::nullopt, 10, 10, 11, 99, 99})
          ->isSorted()
          .value());
  EXPECT_FALSE(
      maker_.dictionary_vector<int64_t>({10, 9, std::nullopt, 10, 10, 10})
          ->isSorted()
          .value());
}
